// Copyright 2011 Edward Wahlen. All Rights Reserved.

package org.wahlen.pc;

import com.google.common.collect.PeekingIterator;

import org.wahlen.io.Marshallers;
import org.wahlen.util.Pair;

import java.io.File;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.Callable;

/**
 * @author edward.wahlen@gmail (Edward Wahlen)
 */
public abstract class PTable<K, V> extends PCollection<Pair<K, V>> {

  private final PType<K> keyType;
  private final PType<V> valueType;

  public PTable(PType<K> keyType, PType<V> valueType) {
    super(PType.pair(keyType, valueType));
    this.keyType = keyType;
    this.valueType = valueType;
  }

  PTable<K, Iterable<V>> groupByKey() {
    return new DiskSortTable<K, V>(this);
  }

  private static class DiskSortTable<K, V> extends PTable<K, Iterable<V>> {

    private final PTable<K, V> pTable;

    public DiskSortTable(PTable<K, V> kvPTable) {
      super(kvPTable.keyType,
          PType.from(Marshallers.oneShotIterable(kvPTable.valueType.marshaller)));
      pTable = kvPTable;
    }

    @Override
    protected Iterator<Pair<K, Iterable<V>>> shard(final int i) {
      // TODO write to NxN grid of sorted collections. Merge on reading. Sort and hash based on byte
      // array encoding so that merging and LogN consolidation need not recreate the actual key
      // value.
      try {
        final String tempPrefix = File.createTempFile("shuffle", "").getName();
        EXECUTOR.invokeAll(sortingWriters(tempPrefix));
        return groupSorted(readSorted(tempPrefix, i));
      } catch (IOException e) {
        throw new AssertionError(e);
      } catch (InterruptedException e) {
        throw new AssertionError(e);
      }
    }

    private Iterator<Pair<K, Iterable<V>>> groupSorted(
        final PeekingIterator<Pair<K, V>> sortedIterator) {
      return new Iterator<Pair<K, Iterable<V>>>() {

        public boolean hasNext() {
          return sortedIterator.hasNext();
        }

        public Pair<K, Iterable<V>> next() {
          if (hasNext()) {
            K currentKey = sortedIterator.peek().first;
            return Pair.of(currentKey, gather(currentKey));
          }
          throw new NoSuchElementException();
        }

        public void remove() {
          throw new UnsupportedOperationException();
        }

        private Iterable<V> gather(final K currentKey) {
          return new Iterable<V>() {
            public Iterator<V> iterator() {
              return new Iterator<V>() {
                public boolean hasNext() {
                  return sortedIterator.hasNext() && sortedIterator.peek().first.equals(currentKey);
                }

                public V next() {
                  if (hasNext()) {
                    return sortedIterator.next().second;
                  }
                  throw new NoSuchElementException();
                }

                public void remove() {
                  throw new UnsupportedOperationException();
                }
              };
            }
          };
        }
      };
    }

    private PeekingIterator<Pair<K, V>> readSorted(String tempPrefix, int i) {
      // TODO
      return null;
    }

    private List<Callable<Void>> sortingWriters(String prefix) {
      // TODO
      return null;
    }
  }
}
